<html>
<head>
  <title>Spark MLlib</title>
  <basefont face="微软雅黑" size="2" />
  <meta http-equiv="Content-Type" content="text/html;charset=utf-8" />
  <meta name="exporter-version" content="Evernote Windows/307027 (zh-CN, DDL); Windows/6.1.0 (Win32);"/>
  <style>
    body, td {
      font-family: 微软雅黑;
      font-size: 10pt;
    }
  </style>
</head>
<body>
<a name="1225"/>
<h1>Spark MLlib</h1>

<div>
<span><div><h2><span style="font-size: 18pt; color: rgb(28, 51, 135);">机器学习</span></h2><div><span style="font-size: 11pt;">该领域的主要研究对象是人工智能。机器学习利用数据或以往的经验，以此优化计算机程序的性能标准。</span></div><div><span style="font-size: 11pt;-en-paragraph:true;">一种经常引用的英文定义是：</span></div><blockquote><div style="margin-top: 1em; margin-bottom: 1em;"><span style="font-size: 11pt;-en-paragraph:true;">A computer program is said to learn from experience E with respect to some class of tasks T and performance measure P, if its performance at tasks in T, as measured by P, improves with experience E。</span></div></blockquote><div><br/></div><div><span style="font-size: 11pt; color: rgb(28, 51, 135);">在大数据上进行机器学习，需要处理全量数据并进行大量的迭代计算，这要求机器学习平台具备强大的处理能力。Spark 立足于内存计算，天然的适应于迭代式计算。即便如此，对于普通开发者来说，实现一个分布式机器学习算法仍然是一件极具挑战的事情。幸运的是，Spark提供了一个基于海量数据的机器学习库，它提供了常用机器学习算法的分布式实现，开发者只需要有 Spark 基础并且了解机器学习算法的原理，以及方法相关参数的含义，就可以轻松的通过调用相应的 API 来实现基于海量数据的机器学习过程。其次，Spark-Shell的即席查询也是一个关键。算法工程师可以边写代码边运行，边看结果。spark提供的各种高效的工具正使得机器学习过程更加直观便捷。比如通过sample函数，可以非常方便的进行抽样。当然，Spark发展到后面，拥有了实时批计算，批处理，算法库，SQL、流计算等模块等，基本可以看做是全平台的系统。把机器学习作为一个模块加入到Spark中，也是大势所趋。</span></div><div><br/></div><div><br/></div><div><br/></div><h2><span style="font-size: 18pt; color: rgb(28, 51, 135);">Spark 机器学习库MLLib</span></h2><div><span style="font-size: 11pt;-en-paragraph:true;">MLlib是Spark的机器学习（Machine Learning）库，旨在简化机器学习的工程实践工作，并方便扩展到更大规模。MLlib由一些通用的学习算法和工具组成，包括分类、回归、聚类、协同过滤、降维等，同时还包括底层的优化原语和高层的管道API。具体来说，其主要包括以下几方面的内容：</span></div><ol><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt; color: rgb(173, 0, 0); font-weight: bold;">算法工具：常用的学习算法，如分类、回归、聚类和协同过滤；</span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt; color: rgb(173, 0, 0); font-weight: bold;">特征化工具：特征提取、转化、降维，和选择工具；</span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt; color: rgb(173, 0, 0); font-weight: bold;">管道(Pipeline)：用于构建、评估和调整机器学习管道的工具;</span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt; color: rgb(173, 0, 0); font-weight: bold;">持久性：保存和加载算法，模型和管道;</span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt; color: rgb(173, 0, 0); font-weight: bold;">实用工具：线性代数，统计，数据处理等工具。</span></span></div></li></ol><div><br/></div><div><span style="font-size: 11pt;-en-paragraph:true;">Spark 机器学习库从 1.2 版本以后被分为两个包：</span></div><ul><li><div><span style="font-size: 11pt;">spark.mllib 包含基于RDD的原始算法API。Spark MLlib 历史比较长，在1.0 以前的版本即已经包含了，提供的算法实现都是基于原始的 RDD。</span></div></li><li><div><a href="http://spark.apache.org/docs/latest/ml-guide.html" style="font-size: 11pt;">spark.ml</a><span style="font-size: 11pt;"> 则提供了基于</span><a href="http://spark.apache.org/docs/latest/sql-programming-guide.html#dataframes" style="font-size: 11pt;">DataFrames</a><span style="font-size: 11pt;"> 高层次的API，可以用来构建机器学习工作流（PipeLine）。ML Pipeline 弥补了原始 MLlib 库的不足，向用户提供了一个基于 DataFrame 的机器学习工作流式 API 套件。</span></div></li></ul><div><br/></div><div><span style="font-size: 11pt;">MLlib目前支持4种常见的机器学习问题: 分类、回归、聚类和协同过滤。下表列出了目前MLlib支持的主要的机器学习算法：</span></div><div><img src="Spark MLlib_files/Image.png" type="image/png" data-filename="Image.png" style="font-size: 11pt;"/></div><div><br/></div><div><span style="font-size: 11pt;">几个重要概念：</span></div><ul><li><div><span style="font-size: 11pt;">DataFrame：使用Spark SQL中的DataFrame作为数据集，它可以容纳各种数据类型。 较之 RDD，包含了 schema 信息，更类似传统数据库中的二维表格。它被 ML Pipeline 用来存储源数据。例如，DataFrame中的列可以是存储的文本，特征向量，真实标签和预测的标签等。</span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt;">Transformer：翻译成转换器，是一种可以将一个DataFrame转换为另一个DataFrame的算法。比如一个模型就是一个 Transformer。它可以把 一个不包含预测标签的测试数据集 DataFrame 打上标签，转化成另一个包含预测标签的 DataFrame。技术上，Transformer实现了一个方法transform（），它通过附加一个或多个列将一个DataFrame转换为另一个DataFrame。</span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt;">Estimator：翻译成估计器或评估器，它是学习算法或在训练数据上的训练方法的概念抽象。在 Pipeline 里通常是被用来操作 DataFrame 数据并生产一个 Transformer。从技术上讲，Estimator实现了一个方法fit（），它接受一个DataFrame并产生一个转换器。如一个随机森林算法就是一个 Estimator，它可以调用fit（），通过训练特征数据而得到一个随机森林模型。</span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt;">Parameter：Parameter 被用来设置 Transformer 或者 Estimator 的参数。现在，所有转换器和估计器可共享用于指定参数的公共API。ParamMap是一组（参数，值）对。</span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt;">PipeLine：翻译为工作流或者管道。工作流将多个工作流阶段（转换器和估计器）连接在一起，形成机器学习的工作流，并获得结果输出。</span></span></div></li></ul><div><br/></div><div><font color="#AD0000" style="font-size: 12pt;"><span style="font-size: 12pt; color: rgb(173, 0, 0); font-weight: bold;">Pipeline工作流: 通常会包含源数据ETL（抽取、转化、加载），数据预处理，指标提取，模型训练与交叉验证，新数据预测等步骤。</span></font></div><div><br/></div><div><span style="font-size: 11pt;">要构建一个 Pipeline工作流，首先需要定义 Pipeline 中的各个工作流阶段PipelineStage，（包括转换器和评估器），比如指标提取和转换模型训练等。有了这些处理特定问题的转换器和 评估器，就可以按照具体的处理逻辑有序的组织PipelineStages 并创建一个Pipeline。比如：</span></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">val pipeline = new Pipeline().setStages(Array(stage1,stage2,stage3,…))</span></div></div><div><div><br/></div><div><br/></div><hr/></div><div><img src="Spark MLlib_files/Image [1].png" type="image/png" data-filename="Image.png" style="font-size: 11pt;"/></div><div><br/></div><ul><li><div><span style="font-size: 11pt;">顶行表示具有三个阶段的流水线。</span> <span style="font-size: 11pt; text-decoration: underline;">前两个（Tokenizer和HashingTF）是Transformers（蓝色）</span><span style="font-size: 11pt;">，</span><span style="font-size: 11pt; color: rgb(173, 0, 0); font-weight: bold;">第三个（LogisticRegression）是Estimator（红色）</span><span style="font-size: 11pt;">。</span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt;">底行表示流经管线的数据，其中圆柱表示DataFrames。 <span style="font-size: 11pt;">在原始DataFrame上调用Pipeline.fit（）方法，它具有原始文本文档和标签。</span></span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt;">Tokenizer.transform（）方法将原始文本文档拆分为单词，向DataFrame添加一个带有单词的新列。 HashingTF.transform（）方法将字列转换为特征向量，向这些向量添加一个新列到DataFrame。</span></span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt;">现在，由于LogisticRegression是一个Estimator，Pipeline首先调用LogisticRegression.fit（）产生一个LogisticRegressionModel。</span></span></div></li></ul><div><br/></div><ul><li><div><span style="font-size: 11pt;">如果流水线有更多的阶段，则在将DataFrame传递到下一个阶段之前，将在DataFrame上调用LogisticRegressionModel的transform（）方法。</span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt;">值得注意的是，工作流本身也可以看做是一个估计器。在工作流的fit（）方法运行之后，它产生一个PipelineModel，它是一个Transformer。</span></span></div></li></ul><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div><font color="#1C3387" style="font-size: 18pt;"><span style="font-size: 18pt; color: rgb(28, 51, 135); font-weight: bold;">工作流（ML Pipelines）例子</span></font></div><div><br/></div><div><font style="font-size: 14pt;"><span style="font-size: 14pt; font-weight: bold;">1、创建名为spark的SparkSession对象</span></font></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><font style="font-size: 14pt;">import org.apache.spark.sql.SparkSession</font></div><div><font style="font-size: 14pt;">val spark = SparkSession.builder().</font></div><div><font style="font-size: 14pt;">            master(&quot;local&quot;).</font></div><div><font style="font-size: 14pt;">            appName(&quot;my App Name&quot;).</font></div><div>            getOrCreate()</div></div><div><span style="font-size: 11pt;">     和 SQLContext一样，也可以开启RDD的隐式转换</span></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div>import spark.implicits._</div></div><div><br/></div><div><br/></div><div><font style="font-size: 14pt;"><span style="font-size: 14pt; font-weight: bold;">2、引入要包含的包并构建训练数据集</span></font></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div>import org.apache.spark.ml.feature._</div><div>import org.apache.spark.ml.classification.LogisticRegression</div><div>import org.apache.spark.ml.{Pipeline,PipelineModel}</div><div>import org.apache.spark.ml.linalg.Vector</div><div>import org.apache.spark.sql.Row</div><div><br/></div><div>scala&gt; val training = spark.createDataFrame(Seq(</div><div>             (0L, &quot;a b c d e spark&quot;, 1.0)</div><div>             (1L, &quot;b d&quot;, 0.0),</div><div>             (2L, &quot;spark f g h&quot;, 1.0),</div><div>             (3L, &quot;hadoop mapreduce&quot;, 0.0)</div><div>            )).toDF(&quot;id&quot;, &quot;text&quot;, &quot;label&quot;)</div></div><div><br/></div><div><br/></div><div><font style="font-size: 14pt;"><span style="font-size: 14pt; font-weight: bold;">3、定义Pipeline中的各个工作流阶段PipelineStage，包括转换器和评估器</span></font></div><div><br/></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div>val tokenizer = new Tokenizer().</div><div>        setInputCol(&quot;text&quot;).</div><div>        setOutputCol(&quot;words&quot;)</div><div>val hashingTF = new HashingTF().</div><div>        setNumFeatures(1000).</div><div>        setInputCol(tokenizer.getOutputCol).</div><div>        setOutputCol(&quot;features&quot;)</div><div>val lr = new LogisticRegression().</div><div>        setMaxIter(10).</div><div>        setRegParam(0.01)</div></div><div><br/></div><div><font style="font-size: 14pt;"><span style="font-size: 14pt; font-weight: bold;">4、按照具体的处理逻辑有序的组织PipelineStages并创建一个Pipeline</span></font></div><div><br/></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div>val pipeline = new Pipeline().</div><div>            setStages(Array(tokenizer, hashingTF, lr))</div><div>val model = pipeline.fit(training)</div><div>model: org.apache.spark.ml.PipelineModel = pipeline_4dabd24db001</div></div><div><span style="font-size: 11pt;">现在构建的Pipeline本质上是一个Estimator，在它的fit（）方法运行之后，它将产生一个PipelineModel，它是一个Transformer。 <span style="font-size: 11pt;">可以看到，model的类型是一个PipelineModel，这个管道模型将在测试数据的时候使用。</span></span></div><div><br/></div><div><br/></div><div><font style="font-size: 14pt;"><span style="font-size: 14pt; font-weight: bold;">5、构建测试数据</span></font></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><span style="font-size: 11pt;"><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">val test = spark.createDataFrame(Seq(</span></span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">          (4L, &quot;spark i j k&quot;),</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">          (5L, &quot;l m n&quot;),</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">          (6L, &quot;spark a&quot;),</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">          (7L, &quot;apache hadoop&quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">        )).toDF(&quot;id&quot;, &quot;text&quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">test: org.apache.spark.sql.DataFrame = [id: bigint, text: string]</span></div></div><div><br/></div><div><br/></div><div><font style="font-size: 14pt;"><span style="font-size: 14pt; font-weight: bold;">6、调用已训练好的PipelineModel的transform()方法，让测试数据按顺序通过拟合的工作流，生成我们所需的预测结果</span></font></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; model.transform(test).</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |       select(&quot;id&quot;, &quot;text&quot;, &quot;probability&quot;, &quot;prediction&quot;).</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |       collect().</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |       foreach { case Row(id: Long, text: String, prob: Vector, prediction: Double) =&gt;</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |         println(s&quot;($id, $text) --&gt; prob=$prob, prediction=$prediction&quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |       }</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">(4, spark i j k) --&gt; prob=[0.5406433544851421,0.45935664551485783], prediction=0.0</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">(5, l m n) --&gt; prob=[0.9334382627383259,0.06656173726167405], prediction=0.0</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">(6, spark a) --&gt; prob=[0.15041430048068286,0.8495856995193171], prediction=1.0</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">(7, apache hadoop) --&gt; prob=[0.9768636139518304,0.023136386048169585], prediction=0.0</span></div></div><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div><font color="#1C3387" style="font-size: 18pt;"><span style="color: rgb(28, 51, 135); font-size: 18pt; font-weight: bold;">特征抽取</span></font></div><div><br/></div><div><font style="font-size: 14pt;"><span style="font-size: 14pt; font-weight: bold;">TF-IDF</span></font></div><div><span style="font-size: 11pt;">“词频－逆向文件频率”（TF-IDF）是一种在文本挖掘中广泛使用的特征向量化方法，它可以体现一个文档中词语在语料库中的重要程度。</span></div><ul><li><div><span style="font-size: 11pt;">词语由t表示，文档由d表示，语料库由D表示。词频TF(t,d)是词语t在文档d中出现的次数。文件频率DF(t,D)是包含词语的文档的个数。</span></div></li><li><div><span style="font-size: 11pt;">如果我们只使用词频来衡量重要性，很容易过度强调在文档中经常出现，却没有太多实际信息的词语，比如“a”，“the”以及“of”。</span></div></li><li><div><span style="font-size: 11pt;">如果一个词语经常出现在语料库中，意味着它并不能很好的对文档进行区分。</span></div></li><li><div><span style="font-size: 11pt;">TF-IDF就是在数值化文档信息，衡量词语能提供多少信息以区分文档。</span></div></li></ul><div><img src="http://dblab.xmu.edu.cn/blog/wp-content/ql-cache/quicklatex.com-1494ed5d67d08b6d7b0c85d5ce8664f6_l3.svg" style="font-size: 11pt;" width="234"></img></div><div><img src="http://dblab.xmu.edu.cn/blog/wp-content/ql-cache/quicklatex.com-347cf1ea57f5023d3a7d83bd5685e4b6_l3.svg" style="font-size: 11pt;" width="21"></img> 是语料库中总的文档数。公式中使用log函数，当词出现在所有文档中时，它的IDF值变为0。加1是为了避免分母为0的情况。</div><div><br/></div><div><img src="http://dblab.xmu.edu.cn/blog/wp-content/ql-cache/quicklatex.com-544dd82127765ba11e0c9f84f3df3285_l3.svg" style="font-size: 11pt;" width="307"></img></div><div><br/></div><div><span style="font-size: 11pt;">在Spark ML库中，TF-IDF被分成两部分：TF (+hashing) 和 IDF。</span></div><ul><li><div><span style="font-size: 11pt; font-weight: bold;">TF</span><span style="font-size: 11pt;">: HashingTF 是一个Transformer，在文本处理中，接收词条的集合然后把这些集合转化成固定长度的特征向量。这个算法在哈希的同时会统计各个词条的词频。</span></div></li><li><div><span style="font-size: 11pt;"><span style="font-size: 11pt; font-weight: bold;">IDF</span><span style="font-size: 11pt;">: IDF是一个Estimator，在一个数据集上应用它的fit（）方法，产生一个IDFModel。 该IDFModel 接收特征向量（由HashingTF产生），然后计算每一个词在文档中出现的频次。IDF会减少那些在语料库中出现频率较高的词的权重。</span></span></div></li></ul><div><br/></div><div><br/></div><div><span style="font-size: 11pt;">例子：</span></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><span style="font-family: Monaco;"><font color="#AD0000" style="font-size: 11pt;"><b>创建一个简单的DataFrame，每一个句子代表一个文档。</b></font></span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val sentenceData = spark.createDataFrame(Seq(</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |       (0, &quot;I heard about Spark and I love Spark&quot;),</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |       (0, &quot;I wish Java could use case classes&quot;),</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |       (1, &quot;Logistic regression models are neat&quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">     |     )).toDF(&quot;label&quot;, &quot;sentence&quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">sentenceData: org.apache.spark.sql.DataFrame = [label: int, sentence: string]</span></div><div><font color="#AD0000" style="font-size: 11pt;"><b><br/></b></font></div><div><span style="font-family: Monaco;"><font color="#AD0000" style="font-size: 11pt;"><b>在得到文档集合后，即可用tokenizer对句子进行分词。</b></font></span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val tokenizer = new Tokenizer().setInputCol(&quot;sentence&quot;).setOutputCol(&quot;words&quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val wordsData = tokenizer.transform(sentenceData)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; wordsData.show(false)</span></div><div>* +-----+------------------------------------+---------------------------------------------+</div><div>* |label|sentence |words |</div><div>* +-----+------------------------------------+---------------------------------------------+</div><div>* |0 |I heard about Spark and I love Spark|[i, heard, about, spark, and, i, love, spark]|</div><div>* |0 |I wish Java could use case classes |[i, wish, java, could, use, case, classes] |</div><div>* |1 |Logistic regression models are neat |[logistic, regression, models, are, neat] |</div><div>* +-----+------------------------------------+---------------------------------------------+</div><div><font color="#AD0000" style="font-size: 11pt;"><b><br/></b></font></div><div><span style="font-family: Monaco;"><font color="#AD0000" style="font-size: 11pt;"><b>使用HashingTF的transform()方法把句子哈希成特征向量，这里设置哈希表的桶数为2000</b></font></span></div><div><span style="font-family: Monaco; font-size: 9pt;">scala&gt; val hashingTF = new HashingTF().</span></div><div><span style="font-family: Monaco; font-size: 9pt;">     |       setInputCol(&quot;words&quot;).setOutputCol(&quot;rawFeatures&quot;).setNumFeatures(2000)</span></div><div><span style="font-family: Monaco; font-size: 9pt;">hashingTF: org.apache.spark.ml.feature.HashingTF = hashingTF_2591ec73cea0</span></div><div><span style="font-family: Monaco; font-size: 9pt;">scala&gt; val featurizedData = hashingTF.transform(wordsData)</span></div><div><span style="font-family: Monaco; font-size: 9pt;">featurizedData: org.apache.spark.sql.DataFrame = [label: int, sentence: string,</span></div><div><span style="font-family: Monaco; font-size: 9pt;">words: array&lt;string&gt;, rawFeatures: vector]</span></div><div><span style="font-family: Monaco; font-size: 9pt;">scala&gt; featurizedData.select(&quot;rawFeatures&quot;).show(false)</span></div><div><span style="font-family: Monaco; font-size: 9pt;">+---------------------------------------------------------------------+</span></div><div><span style="font-family: Monaco; font-size: 9pt;">|rawFeatures                                                          |</span></div><div><span style="font-family: Monaco; font-size: 9pt;">+---------------------------------------------------------------------+</span></div><div><span style="font-family: Monaco; font-size: 9pt;">|(2000,[240,333,1105,1329,1357,1777],[1.0,1.0,2.0,2.0,1.0,1.0])       |</span></div><div><span style="font-family: Monaco; font-size: 9pt;">|(2000,[213,342,489,495,1329,1809,1967],[1.0,1.0,1.0,1.0,1.0,1.0,1.0])|</span></div><div><span style="font-family: Monaco; font-size: 9pt;">|(2000,[286,695,1138,1193,1604],[1.0,1.0,1.0,1.0,1.0])                |</span></div><div><span style="font-family: Monaco; font-size: 9pt;">+---------------------------------------------------------------------+</span></div><div><span style="font-family: Monaco; font-size: 9pt;"><br/></span></div><div><span style="font-family: Monaco;"><font color="#AD0000" style="font-size: 11pt;"><b><br/></b></font></span></div><div><span style="font-family: Monaco;"><font color="#AD0000" style="font-size: 11pt;"><b>最后，使用IDF来对单纯的词频特征向量进行修正，使其更能体现不同词汇对文本的区别能力，IDF是一个Estimator，调用fit()方法并将词频向量传入，即产生一个IDFModel</b></font></span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val idf = new IDF().setInputCol(&quot;rawFeatures&quot;).setOutputCol(&quot;features&quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">idf: org.apache.spark.ml.feature.IDF = idf_7fcc9063de6f</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val idfModel = idf.fit(featurizedData)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">idfModel: org.apache.spark.ml.feature.IDFModel = idf_7fcc9063de6f</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);"><br/></span></div><div><span style="font-family: Monaco;"><font color="#AD0000" style="font-size: 11pt;"><b>很显然，IDFModel是一个Transformer，调用它的transform()方法，即可得到每一个单词对应的TF-IDF度量值</b></font></span></div><div><span style="font-family: Monaco;"><font color="#AD0000" style="font-size: 11pt;"><b><br/></b></font></span></div><div><span style="font-family: Monaco; font-size: 9pt;">scala&gt; val rescaledData = idfModel.transform(featurizedData)</span></div><div><span style="font-family: Monaco; font-size: 9pt;">rescaledData: org.apache.spark.sql.DataFrame = [label: int, sentence: string, words: array&lt;string&gt;, rawFeatures: vector, features: vector]</span></div><div><span style="font-family: Monaco; font-size: 9pt;">scala&gt; rescaledData.select(&quot;features&quot;, &quot;label&quot;).take(3).foreach(println)</span></div><div><span style="font-family: Monaco; font-size: 9pt;">[(2000,[240,333,1105,1329,1357,1777],[0.6931471805599453,0.6931471805599453,1.3862943611198906,0.5753641449035617,0.6931471805599453,0.6931471805599453]),0]</span></div><div><span style="font-family: Monaco; font-size: 9pt;">[(2000,[213,342,489,495,1329,1809,1967],[0.6931471805599453,0.6931471805599453,0.6931471805599453,0.6931471805599453,0.28768207245178085,0.6931471805599453,0.6931471805599453]),0]</span></div><div><span style="font-family: Monaco; font-size: 9pt;">[(2000,[286,695,1138,1193,1604],[0.6931471805599453,0.6931471805599453,0.6931471805599453,0.6931471805599453,0.6931471805599453]),1]</span></div></div><div><br/></div><div><br/></div><div><br/></div><div><font style="font-size: 14pt;"><b>Word2Vec</b></font></div></div><div><img src="Spark MLlib_files/Image [2].png" type="image/png" data-filename="Image.png" style="font-size: 11pt;" width="253"/> <img src="Spark MLlib_files/Image [3].png" type="image/png" data-filename="Image.png" width="270"/></div><div><span style="font-size: 11pt;-en-paragraph:true;">在</span><span style="font-size: 11pt;-en-paragraph:true;">ml</span><span style="font-size: 11pt;-en-paragraph:true;">库中，Word2vec 的实现使用的是skip-gram模型。Skip-gram的训练目标是学习词表征向量分布，其优化目标是在给定中心词的词向量的情况下，最大化以下似然函数：</span></div><div><span style="line-height: 57px; font-size: 11pt;-en-paragraph:true;"> </span> <span style="line-height: 57px; font-size: 11pt;-en-paragraph:true;"> </span> <img src="http://dblab.xmu.edu.cn/blog/wp-content/ql-cache/quicklatex.com-9a8cb93eb8c1f98ebd5c1ae98b8b3390_l3.svg" style="font-size: 11pt;" width="189"></img></div><div><span style="font-size: 11pt;-en-paragraph:true;">分别代表当前词的词向量以及当前上下文的词向量表示：</span></div><div><span style="line-height: 51px; font-size: 11pt;-en-paragraph:true;"> </span> <span style="line-height: 51px; font-size: 11pt;-en-paragraph:true;"> </span><span style="line-height: 51px; font-size: 11pt;-en-paragraph:true;"> </span><img src="http://dblab.xmu.edu.cn/blog/wp-content/ql-cache/quicklatex.com-629a40c8811841ca04b8318081b0fee0_l3.svg" style="font-size: 11pt;" width="225"></img></div><div><font style="font-size: 11pt;"><span style="font-size: 11pt;-en-paragraph:true;">因为Skip-gram模型使用的softmax计算较为复杂，所以，</span><span style="font-size: 11pt;-en-paragraph:true;">ml</span><span style="font-size: 11pt;-en-paragraph:true;">与其他经典的Word2Vec实现采用了相同的策略，使用Huffman树来进行</span> <span style="font-weight: bold; font-size: 11pt;-en-paragraph:true;">层次Softmax(Hierachical Softmax)</span> <span style="font-size: 11pt;-en-paragraph:true;">方法来进行优化，使得</span></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><span style="line-height: 19px; font-size: 11pt;-en-paragraph:true;"> </span> <span style="line-height: 19px; font-size: 11pt;-en-paragraph:true;"> </span> <img src="http://dblab.xmu.edu.cn/blog/wp-content/ql-cache/quicklatex.com-97fe92ae497854f3fff0314b8baa30ab_l3.svg" style="font-size: 11pt;" width="91"></img><span style="font-size: 11pt;">计算的复杂度从</span><span style="line-height: 18px; font-size: 11pt;-en-paragraph:true;"> </span><span style="line-height: 18px; font-size: 11pt;-en-paragraph:true;"> </span> <img src="http://dblab.xmu.edu.cn/blog/wp-content/ql-cache/quicklatex.com-ec960b1885f33c255f8efabad9bbfa15_l3.svg" style="font-size: 11pt;" width="41"></img><span style="font-size: 11pt;">下降到</span><span style="line-height: 18px; font-size: 11pt;-en-paragraph:true;"> </span> <img src="http://dblab.xmu.edu.cn/blog/wp-content/ql-cache/quicklatex.com-e8eb0b8c522cf861ec07c1b15ff6e105_l3.svg" style="font-size: 11pt;" width="79"></img></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><br/></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 12pt; color: rgb(173, 0, 0);"><b>在下面的代码段中，我们首先用一组文档，其中一个词语序列代表一个文档。对于每一个文档，我们将其转换为一个特征向量。此特征向量可以被传递到一个学习算法。</b></font><br/></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;">1、导入Word2Vec所需要的包，并创建三个词语序列，每个代表一个文档</font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">import org.apache.spark.ml.feature.Word2Vec</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val documentDF = spark.createDataFrame(Seq(</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">|       &quot;Hi I heard about Spark&quot;.split(&quot; &quot;),</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">|       &quot;I wish Java could use case classes&quot;.split(&quot; &quot;),</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">|       &quot;Logistic regression models are neat&quot;.split(&quot; &quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">|     ).map(Tuple1.apply)).toDF(&quot;text&quot;)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">documentDF: org.apache.spark.sql.DataFrame = [text: array&lt;string&gt;]</span></div></div><br/></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;">2、新建一个Word2Vec，显然他是一个Estimator，设置相应的超参数，这里设置向量的维度为3，Word2Vec模型还有其他可设置的超参数</font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val word2Vec = new Word2Vec().</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">|       setInputCol(&quot;text&quot;).</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">|       setOutputCol(&quot;result&quot;).</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">|       setVectorSize(3).</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">|       setMinCount(0)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">word2Vec: org.apache.spark.ml.feature.Word2Vec = w2v_e2d5128ba199</span></div></div><br/></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;">3、读入训练数据，用fit(）方法生成一个Word2VecModel</font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val model = word2Vec.fit(documentDF)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">model: org.apache.spark.ml.feature.Word2VecModel = w2v_e2d5128ba199</span></div></div><br/></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;">4、利用Word2VecModel把文档转变成特征向量</font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; val result = model.transform(documentDF)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">result: org.apache.spark.sql.DataFrame = [text: array&lt;string&gt;, result: vector]</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">scala&gt; result.select(&quot;result&quot;).take(3).foreach(println)</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">[[0.018490654602646827,-0.016248732805252075,0.04528368394821883]]</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">[[0.05958533100783825,0.023424440695505054,-0.027310076036623544]]</span></div><div><span style="font-family: Monaco; font-size: 9pt; color: rgb(51, 51, 51);">[[-0.011055880039930344,0.020988055132329465,0.042608972638845444]]</span></div></div><br/></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 14pt;"><b>CountVectorizer</b></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><span style="font-size: 11pt;-en-paragraph:true;">CountVectorizer</span><span style="font-size: 11pt;-en-paragraph:true;">旨在通过计数来将一个文档转换为向量。当不存在先验字典时，</span><span style="font-size: 11pt;-en-paragraph:true;">Countvectorizer</span><span style="font-size: 11pt;-en-paragraph:true;">作为</span><span style="font-size: 11pt;-en-paragraph:true;">Estimator</span><span style="font-size: 11pt;-en-paragraph:true;">提取词汇进行训练，并生成一个</span><span style="font-size: 11pt;-en-paragraph:true;">CountVectorizerModel</span><span style="font-size: 11pt;-en-paragraph:true;">用于存储相应的词汇向量空间。该模型产生文档关于词语的稀疏表示，其表示可以传递给其他算法，例如LDA。</span></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><span style="font-size: 11pt;-en-paragraph:true;">在</span><span style="font-size: 11pt;-en-paragraph:true;">CountVectorizerModel</span><span style="font-size: 11pt;-en-paragraph:true;">的训练过程中，</span><span style="font-size: 11pt;-en-paragraph:true;">CountVectorizer</span><span style="font-size: 11pt;-en-paragraph:true;">将根据语料库中的词频排序从高到低进行选择，词汇表的最大含量由</span><span style="font-size: 11pt;-en-paragraph:true;">vocabsize</span><span style="font-size: 11pt;-en-paragraph:true;">超参数来指定，超参数</span><span style="font-size: 11pt;-en-paragraph:true;">minDF</span><span style="font-size: 11pt;-en-paragraph:true;">，则指定词汇表中的词语至少要在多少个不同文档中出现。</span><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;">程序：</font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div>import org.apache.spark.ml.feature.{CountVectorizer, CountVectorizerModel}</div><div> val df = spark.createDataFrame(Seq(</div><div> (0, Array(&quot;a&quot;, &quot;b&quot;, &quot;c&quot;)),</div><div> (1, Array(&quot;a&quot;, &quot;b&quot;, &quot;b&quot;, &quot;c&quot;, &quot;a&quot;))</div><div> )).toDF(&quot;id&quot;, &quot;words&quot;)</div><div><br/></div><div>val cvModel: CountVectorizerModel = new CountVectorizer().</div><div> setInputCol(&quot;words&quot;).</div><div> setOutputCol(&quot;features&quot;).</div><div> setVocabSize(3).</div><div> setMinDF(2).</div><div> fit(df)</div><div><br/></div><div>val cvm = new CountVectorizerModel(Array(&quot;a&quot;, &quot;b&quot;, &quot;c&quot;)).</div><div> setInputCol(&quot;words&quot;).</div><div> setOutputCol(&quot;features&quot;)</div></div><br/></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font><br/></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div style="line-height: 19px; margin-top: 1em; margin-bottom: 1em;"><font style="font-size: 11pt;"><br/></font></div><div><br/></div></span>
</div></body></html> 